package lac.serviceuint;

import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Random;
import java.util.UUID;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.Watcher.Event.EventType;
import org.apache.zookeeper.Watcher.Event.KeeperState;
import org.apache.zookeeper.ZKUtil;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooDefs.Ids;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.ACL;

import com.fasterxml.jackson.databind.DeserializationFeature;
import com.fasterxml.jackson.databind.ObjectMapper;

import lac.serviceuint.register.RegistryClient;
import lac.serviceuint.register.RegistryHost;
import lac.serviceuint.register.RegistryService;
import lac.serviceuint.register.RegistryServiceMethod;

public class ServiceUnitRegister {

	private static Log log = LogFactory.getLog(ServiceUnitRegister.class);
	
	private Map<String,RegistryService> registryServiceCache = new LinkedHashMap<>();
	private Map<String,RegistryServiceMethod> registryServiceMethodCache = new LinkedHashMap<>();
	private ObjectMapper mapper = new ObjectMapper();
	
	private long hostExpiredTime = 1000*30;
	private Thread clearExpiredHostThread = null;
	private long leaderExpiredTime = 1000*60;
	private Boolean applyLeader = true;
	private Thread applyLeaderThread =  new Thread(new ApplyLeaderThread(),"ServiceUnitRegister-ApplyLeaderThread");
	
	private long refreshTime = 1000*10;
	
	
	private ZooKeeper zooKeeper;
	private Properties properties;
	private List<ACL> acl = new ArrayList<ACL>();
	private String registerId = UUID.randomUUID().toString();
	public enum keys {
		zkConnectString, zkPath,zkSessionTimeout,refreshTime,hostExpiredTime,applyLeader
	}
	
	private  Thread createClearExpiredHostThread(){
		clearExpiredHostThread =  new Thread(new ClearExpiredHostThread(),"ServiceUnitRegister-ClearExpiredHostThread");
		return clearExpiredHostThread;
	}
	
	public ServiceUnitRegister(Properties properties) throws Exception {
		this.properties = properties;
		if(properties.get(keys.refreshTime.name())!=null){
			refreshTime = Long.parseLong(this.properties.getProperty(keys.refreshTime.name()));
		}
		if(properties.get(keys.hostExpiredTime.name())!=null){
			hostExpiredTime = Long.parseLong(this.properties.getProperty(keys.hostExpiredTime.name()));
		}
		if(properties.get(keys.applyLeader.name())!=null){
			applyLeader = Boolean.valueOf(this.properties.getProperty(keys.applyLeader.name()));
		}
		mapper.disable(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES);
		//mapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
		connect();
	}
	
	public void init(){
		registryServiceCache.clear();
		registryServiceMethodCache.clear();
		try{
			NodeWatcher nodeWatcher = new NodeWatcher(zooKeeper);
			if(zooKeeper.exists(getRootPath(), null)==null){
				Map<String,Object> leaderInfo = new HashMap<>();
				leaderInfo.put("registerId", registerId);
				leaderInfo.put("lastUpdateTime", System.currentTimeMillis());
				zooKeeper.create(getRootPath(), mapper.writeValueAsBytes(leaderInfo), acl, CreateMode.PERSISTENT);
				zooKeeper.getChildren(getRootPath(),nodeWatcher);
			}else{
				List<String> pathList = ZKUtil.listSubTreeBFS(zooKeeper,getRootPath());
				for(String path : pathList){
					zooKeeper.getChildren(path,nodeWatcher);
					updateCache(path,nodeWatcher);
				}
			}
		}catch(Exception e){
			e.printStackTrace();
		}
		if(applyLeader){
			applyLeaderThread.start();
		}
	}
	
	private void addNodeWatcher(String parentPath,List<String> paths,NodeWatcher nodeWatcher) throws Exception{
		List<String> watcherPaths =  new ArrayList<>();
		for(String path:paths){
			path=parentPath+"/"+path;
			if(path.split("/").length==3 && registryServiceCache.get(path)==null){
				watcherPaths.add(path);
				updateCache(path,nodeWatcher);
			}else if((path.split("/").length==4) && registryServiceMethodCache.get(path)==null){
				watcherPaths.add(path);
				updateCache(path,nodeWatcher);
			}
		}
		for(String path:watcherPaths){
			zooKeeper.getChildren(path, nodeWatcher);
		}
	}
	
	private void updateCache(String path,NodeWatcher nodeWatcher) throws Exception{
		byte[] data = zooKeeper.getData(path, nodeWatcher, null);
		if(path.split("/").length==3){
			RegistryService service = mapper.readValue(data, RegistryService.class);
			registryServiceCache.put(path, service);
		}else if((path.split("/").length==4)){
			RegistryServiceMethod serviceUnitMethod = mapper.readValue(data, RegistryServiceMethod.class);
			registryServiceMethodCache.put(path, serviceUnitMethod);
		}
	}
	
	
	public void addRegisterService(RegistryService registryService) throws Exception{
		String serviceId = registryService.getServiceId();
		String serviceIdPath = getRootPath()+"/"+serviceId;
		
		if(zooKeeper.exists(serviceIdPath, null)!=null){
			throw new RuntimeException("服务"+serviceId+"已存在！");
		}
		else{
			zooKeeper.create(serviceIdPath, mapper.writeValueAsBytes(registryService), acl, CreateMode.PERSISTENT);
		}
	}
	
	public void updateRegisterService(RegistryService registryService) throws Exception{
		
		String serviceId = registryService.getServiceId();
		String serviceIdPath = getRootPath()+"/"+serviceId;
		if(zooKeeper.exists(serviceIdPath, null)==null){
			throw new RuntimeException("服务"+serviceId+"不存在！");
		}else{
			zooKeeper.setData(serviceIdPath, mapper.writeValueAsBytes(registryService), -1);
		}
	}
	
	public void deleteRegisterService(String serviceId) throws Exception{
		ZKUtil.deleteRecursive(zooKeeper, getRootPath()+"/"+serviceId);
	}
	
	public void addRegistryServiceMethod(RegistryServiceMethod registryServiceMethod) throws Exception{
		String serviceId = registryServiceMethod.getServiceId();
		String serviceIdPath = getRootPath()+"/"+serviceId;
		if(zooKeeper.exists(serviceIdPath, null)==null){
			throw new RuntimeException("服务"+serviceId+"不存在！");
		}else{
			String serviceUnitName = registryServiceMethod.getServiceUnitName()==null?"":registryServiceMethod.getServiceUnitName();
			String serviceUnitMethodName = registryServiceMethod.getMethodName();
			String serviceUnitMethodPath = serviceIdPath+"/"+serviceUnitName+"_"+serviceUnitMethodName;
			if(zooKeeper.exists(serviceUnitMethodPath, null)!=null){
				throw new RuntimeException("服务"+serviceId+"的方法"+serviceUnitName+"_"+serviceUnitMethodName+"已存在！");
			}else{
				zooKeeper.create(serviceUnitMethodPath, mapper.writeValueAsBytes(registryServiceMethod), acl, CreateMode.PERSISTENT);
			}
		}
	}
	
	public void updateRegistryServiceMethod(RegistryServiceMethod registryServiceMethod) throws Exception{
		String serviceId = registryServiceMethod.getServiceId();
		String serviceIdPath = getRootPath()+"/"+serviceId;
		if(zooKeeper.exists(serviceIdPath, null)==null){
			throw new RuntimeException("服务"+serviceId+"不存在！");
		}else{
			String serviceUnitName = registryServiceMethod.getServiceUnitName()==null?"":registryServiceMethod.getServiceUnitName();
			String serviceUnitMethodName = registryServiceMethod.getMethodName();
			String serviceUnitMethodPath = serviceIdPath+"/"+serviceUnitName+"_"+serviceUnitMethodName;
			if(zooKeeper.exists(serviceUnitMethodPath, null)==null){
				throw new RuntimeException("服务"+serviceId+"的方法"+serviceUnitName+"_"+serviceUnitMethodName+"不存在！");
			}else{
				zooKeeper.setData(serviceUnitMethodPath, mapper.writeValueAsBytes(registryServiceMethod),-1);
			}
		}
	}
	
	public void deleteRegistryServiceMethod(RegistryServiceMethod registryServiceMethod) throws Exception{
		zooKeeper.delete(getRootPath()+"/"+registryServiceMethod.getServiceId()+"/"+registryServiceMethod.getServiceUnitName()+"_"+registryServiceMethod.getMethodName(), -1);
	}
	
	public List<RegistryService> getRegistryService(){
		List<RegistryService> registryServices = new ArrayList<>();
		registryServices.addAll(registryServiceCache.values());
		return registryServices;
	}
	
	public List<RegistryServiceMethod> getRegistryServiceMethod(String serviceId){
		List<RegistryServiceMethod> registryServiceMethods = new ArrayList<>();
		for(RegistryServiceMethod method:registryServiceMethodCache.values()){
			if(method.getServiceId().equals(serviceId)){
				registryServiceMethods.add(method);
			}
		}
		return registryServiceMethods;
	}
	
	public List<String> getMethodHostUrls(String serviceId,String serviceUnitName,String methodName,String group){
		String methodPath = getRootPath()+"/"+serviceId+"/"+serviceUnitName+"_"+methodName;
		RegistryServiceMethod registryServiceMethod =registryServiceMethodCache.get(methodPath);
		List<String> list = new ArrayList<>();
		if(registryServiceMethod != null && registryServiceMethod.getRegistryHosts() != null){
			List<RegistryHost> groupRegistryHostList = new ArrayList<>();
			if(group == null){
				groupRegistryHostList.addAll(registryServiceMethod.getRegistryHosts());
			}else{
				for(RegistryHost registryHost : registryServiceMethod.getRegistryHosts()){
					if(registryHost.getGroup().equals(group)){
						groupRegistryHostList.add(registryHost);
					}
				}
			}
			for(RegistryHost host : groupRegistryHostList){
				list.add(host.getHost()+registryServiceMethod.getUri());
			}
		}
		return list;
	}
	
	public String getMethodHostUrl(String serviceId,String serviceUnitName,String methodName,String group,String clientId) throws Exception{
		String methodPath = getRootPath()+"/"+serviceId+"/"+serviceUnitName+"_"+methodName;
		RegistryServiceMethod registryServiceMethod =registryServiceMethodCache.get(methodPath);
		if(registryServiceMethod==null){
			throw new RuntimeException("服务"+serviceId+"的方法"+serviceUnitName+"_"+methodName+"不存在！");
		}
		if(clientId != null){
			RegistryClient registryClient = new RegistryClient();
			registryClient.setClientId(clientId);
			registryClient.setLastUpdateTime(new Date().getTime());
			List<RegistryClient> registryClients = registryServiceMethod.getRegistryClients();
			boolean isContainClient = false;
			for(RegistryClient cli : registryClients){
				if(cli.getClientId()!=null && cli.getClientId().equals(registryClient.getClientId())){
					cli.setLastUpdateTime(new Date().getTime());
					isContainClient=true;
					break;
				}
			}
			if(!isContainClient){
				registryClients.add(registryClient);
			}
			zooKeeper.setData(methodPath, mapper.writeValueAsBytes(registryServiceMethod), -1);
		}
		
		if(registryServiceMethod.getRegistryHosts()!=null && registryServiceMethod.getRegistryHosts().size()>0){
			List<RegistryHost> groupRegistryHostList = new ArrayList<>();
			if(group == null){
				groupRegistryHostList.addAll(registryServiceMethod.getRegistryHosts());
			}else{
				for(RegistryHost registryHost : registryServiceMethod.getRegistryHosts()){
					if(registryHost.getGroup()!=null && registryHost.getGroup().equals(group)){
						groupRegistryHostList.add(registryHost);
					}
				}
			}
			if(groupRegistryHostList.size()>0){
				Random random = new Random();
				int index = random.nextInt(groupRegistryHostList.size());
				RegistryHost host = groupRegistryHostList.get(index);
				return host.getHost()+registryServiceMethod.getUri();
			}
			
		}
		return null;
	}
	
	public void registryServiceMethodHeartBeat(String serviceId,String serviceUnitName,String serviceUnitMethodName,String group,String host,int port) throws Exception{
		String serviceIdPath = getRootPath()+"/"+serviceId;
		String serviceUnitMethodPath = serviceIdPath+"/"+serviceUnitName+"_"+serviceUnitMethodName;
		RegistryHost registryHost = new RegistryHost();
		registryHost.setHost(host+":"+port);
		registryHost.setGroup(group);
		List<RegistryHost> registryHosts = null;
		if(zooKeeper.exists(serviceIdPath, null)==null){
			throw new RuntimeException("服务"+serviceId+"不存在！");
		}else{
			byte[] serviceData = zooKeeper.getData(serviceIdPath, null, null);
			RegistryService service = mapper.readValue(serviceData, RegistryService.class);
			if(service.getRegistryHosts()==null){
				registryHosts =  new ArrayList<>();
				registryHosts.add(registryHost);
				service.setRegistryHosts(registryHosts);
			}else{
				registryHosts = service.getRegistryHosts();
				boolean isContainHost = false;
				for(RegistryHost rh : registryHosts){
					if(rh.getHost().equals(registryHost.getHost())){
						rh.setLastUpdateTime(new Date().getTime());
						isContainHost=true;
						break;
					}
				}
				if(!isContainHost){
					registryHosts.add(registryHost);
				}
			}
			zooKeeper.setData(serviceIdPath, mapper.writeValueAsBytes(service), -1);
		}
		
		if(zooKeeper.exists(serviceUnitMethodPath, null)==null){
			throw new RuntimeException("服务"+serviceId+"中没有注册"+serviceUnitName+"_"+serviceUnitMethodName+"方法！");
		}else{
			byte[] methodData = zooKeeper.getData(serviceUnitMethodPath, null, null);
			RegistryServiceMethod serviceUnitMethod = mapper.readValue(methodData, RegistryServiceMethod.class);
			if(serviceUnitMethod.getRegistryHosts()==null){
				registryHosts =  new ArrayList<>();
				registryHosts.add(registryHost);
				serviceUnitMethod.setRegistryHosts(registryHosts);
			}else{
				registryHosts = serviceUnitMethod.getRegistryHosts();
				boolean isContainHost = false;
				for(RegistryHost rh : registryHosts){
					if(rh.getHost().equals(registryHost.getHost())){
						isContainHost=true;
						rh.setLastUpdateTime(new Date().getTime());
						rh.setGroup(group);
						break;
					}
				}
				if(!isContainHost){
					registryHosts.add(registryHost);
				}
			}
			zooKeeper.setData(serviceUnitMethodPath, mapper.writeValueAsBytes(serviceUnitMethod), -1);
		}
	}
	
	private String getRootPath(){
		return this.properties.getProperty(keys.zkPath.toString());
	}
	
	private synchronized void  reConnection() throws Exception{
		if (this.zooKeeper != null) {
			this.zooKeeper.close();
			this.zooKeeper = null;
			this.connect() ;
		}
	}
	
	private void connect() throws Exception {
		zooKeeper = new ZooKeeper(
				  this.properties.getProperty(keys.zkConnectString.toString())
				, Integer.parseInt(this.properties.getProperty(keys.zkSessionTimeout.toString()))
				,new Watcher() {
					public void process(WatchedEvent event) {
						if (event.getState() == KeeperState.SyncConnected) {
							log.info("收到ZK连接成功事件！");
							log.info("注册中心id："+registerId);
							init();
						} else if (event.getState() == KeeperState.Expired ) {
							log.error("会话超时，等待重新建立ZK连接...");
							try {
								reConnection();
							} catch (Exception e) {
								log.error(e.getMessage(),e);
							}
						}
						else if (event.getState() == KeeperState.Disconnected ) {
							log.info("等待重新建立ZK连接...");
							try {
								reConnection();
							} catch (Exception e) {
								log.error(e.getMessage(),e);
							}
						}
						else{
							log.info("会话有其他状态的值，event.getState() ="+event.getState() +", event  value="+event.toString());
						}
					}
				});
		acl.add(new ACL(ZooDefs.Perms.ALL,Ids.ANYONE_ID_UNSAFE));
	}
	
	
	class NodeWatcher implements Watcher{
		
		private ZooKeeper zooKeeper;
		
		public NodeWatcher(ZooKeeper zooKeeper){
			this.zooKeeper = zooKeeper;
		}
		@Override
		public void process(WatchedEvent event) {
			try {
				
				if(EventType.NodeChildrenChanged.equals(event.getType())){
					if(zooKeeper.exists(event.getPath(), null)!=null){
						List<String> childrenPaths = zooKeeper.getChildren(event.getPath(), this);
						addNodeWatcher(event.getPath(),childrenPaths,this);
					}
				}else if(EventType.NodeDataChanged.equals(event.getType())){
					//zooKeeper.getData(event.getPath(), this, null);
					updateCache(event.getPath(),this);
				}else if(EventType.NodeDeleted.equals(event.getType())){
					registryServiceCache.remove(event.getPath());
					registryServiceMethodCache.remove(event.getPath());
				}
			}catch (Exception e) {
				e.printStackTrace();
			}
			
		}
	}
	
	
	class ApplyLeaderThread implements Runnable{
		@Override
		public void run() {
			while(true){
				try {
					Thread.sleep(refreshTime);
					byte[] data = zooKeeper.getData(getRootPath(), null,null);
					Map<String,Object> leaderInfo=new HashMap<>();
					String leaderId="";
					Long lastUpdateTime=0L;
					if(data != null){
						try{
							leaderInfo = mapper.readValue(data, Map.class);
							leaderId = leaderInfo.get("registerId").toString();
							lastUpdateTime = (Long) leaderInfo.get("lastUpdateTime");
							log.debug("leader info："+mapper.writeValueAsString(leaderInfo));
						}catch(Exception e){
							leaderInfo =  new HashMap<>();
						}
					}
					leaderInfo.put("registerId", registerId);
					leaderInfo.put("lastUpdateTime", System.currentTimeMillis());
					if(leaderId.equals(registerId)){
						log.debug("Self is leader. id："+registerId);
						if(clearExpiredHostThread == null){
							clearExpiredHostThread = createClearExpiredHostThread();
							clearExpiredHostThread.start();
						}
						zooKeeper.setData(getRootPath(), mapper.writeValueAsBytes(leaderInfo), -1);
					}else{
						if(clearExpiredHostThread != null){
							clearExpiredHostThread.interrupt();
						}
						log.debug("System time:"+System.currentTimeMillis()+" leader lastUpdateTime:"+lastUpdateTime+" The difference:"+String.valueOf(System.currentTimeMillis()-lastUpdateTime));
						if((System.currentTimeMillis()-lastUpdateTime)>leaderExpiredTime){
							log.debug("leader is expired ,try to apply leader. id："+registerId);
							zooKeeper.setData(getRootPath(), mapper.writeValueAsBytes(leaderInfo), -1);
						}
					}
				} catch (InterruptedException e) {
					e.printStackTrace();
				} catch (Exception e) {
					e.printStackTrace();
				}
				
				
			}
			
		}
		
	}
	
	class ClearExpiredHostThread implements Runnable{
		@Override
		public void run() {
			while(true){
				try {
					log.info("服务："+mapper.writeValueAsString(registryServiceCache));
					log.info("服务方法："+mapper.writeValueAsString(registryServiceMethodCache));
					Thread.sleep(refreshTime);
					List<RegistryHost> registryHosts = null;
					Long lastUpdateTime=null;
					
					for(RegistryService registryService : registryServiceCache.values()){
						registryHosts = registryService.getRegistryHosts();
						if(registryService.getRegistryHosts()!=null){
							Iterator<RegistryHost> iterator = registryHosts.iterator();
							while(iterator.hasNext()){
								RegistryHost registryHost = iterator.next();
								lastUpdateTime = registryHost.getLastUpdateTime();
								if(lastUpdateTime==null || ((System.currentTimeMillis()-lastUpdateTime)>hostExpiredTime)){
									iterator.remove();
									updateRegisterService(registryService);
								}
							}
						}
					}
					
					for(RegistryServiceMethod registryServiceMethod : registryServiceMethodCache.values()){
						registryHosts = registryServiceMethod.getRegistryHosts();
						if(registryServiceMethod.getRegistryHosts()!=null){
							Iterator<RegistryHost> iterator = registryHosts.iterator();
							while(iterator.hasNext()){
								RegistryHost registryHost = iterator.next();
								lastUpdateTime = registryHost.getLastUpdateTime();
								if(lastUpdateTime==null || ((System.currentTimeMillis()-lastUpdateTime)>hostExpiredTime)){
									iterator.remove();
									updateRegistryServiceMethod(registryServiceMethod);
								}
							}
						}
					}
					
				} catch (InterruptedException e) {
					e.printStackTrace();
					break;
				} 
				catch (Exception e) {
					e.printStackTrace();
				}
			}
			
		}
		
	}
	
}





